package com.twoandone.tms.realtime.utils;

import com.twoandone.tms.realtime.beans.TransientSink;
import com.twoandone.tms.realtime.common.TmsConfig;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

/**
 * @packageName:com.twoandone.tms.realtime.utils
 * @ClassName:MyClickHouseSinkFunction
 * @Description:
 * @author:
 * @date 2023/3/11 20:18
 **/
public class MyClickHouseSinkFunction {

    public static <T> SinkFunction<T> getSinkFunction(String sql) {
        SinkFunction<T> sink = JdbcSink.sink(sql, new JdbcStatementBuilder<T>() {
                    // insert into dws_traffic_keyword_page_view_window values(?,?,?,?,?)
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {

                        Field[] fields = t.getClass().getDeclaredFields();
                        int num = 0;
                        for (int i = 0; i < fields.length; i++) {
                            Field field = fields[i];
                            TransientSink annotation = field.getAnnotation(TransientSink.class);
                            if (annotation != null) {
                                num++;
                                continue;
                            }

                            field.setAccessible(true);
                            try {
                                Object value = field.get(t);
                                preparedStatement.setObject(i + 1 - num, value);
                            } catch (IllegalAccessException e) {
                                e.printStackTrace();
                            }
                        }
                    }
                },
                new JdbcExecutionOptions.Builder()
                        .withBatchSize(10)
                        .withBatchIntervalMs(5000L)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(TmsConfig.CLICKHOUSE_DRIVER)
                        .withUrl(TmsConfig.CLICKHOUSE_URL)
                        .build()
        );
        return sink;
    }
}
